package com.erbao.nettystudy.nio;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * nio的多线程模型:
 * 相当于一个线程门口迎客，另外的线程服务客户端连接
 * @author CaoBaoYin
 * @since 2021/10/23 19:46
 */
public class PoolServer {
    /*  */
    ExecutorService pool = Executors.newFixedThreadPool(50);
    private Selector selector;
    public static void main(String[] args) throws IOException {
        PoolServer server = new PoolServer();
        server.initServer (8888);
        server.listen();
    }

    private void initServer(int port) throws IOException {
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.socket().bind(new InetSocketAddress(8888));
        ssc.configureBlocking(false);//默认阻塞，设为非阻塞形式

        selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);

        System.out.println("服务端启动成功");
    }

    private void listen() throws IOException {
        while (true) {
            selector.select();

            Iterator<SelectionKey> it = this.selector.selectedKeys().iterator();
            while (it.hasNext()) {
                SelectionKey key = it.next();

                it.remove();

                if (key.isAcceptable()) {
                    ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    socketChannel.register(this.selector, SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    // 把OP_READ剔除
                    key.interestOps(key.interestOps() & (~SelectionKey.OP_READ));

                    pool.execute(new ChannelHandlerThread(key));
                }
            }
        }
    }
}

class ChannelHandlerThread implements Runnable {
    private SelectionKey key;

    public ChannelHandlerThread(SelectionKey key) {
        this.key = key;
    }

    @Override
    public void run() {
        SocketChannel socketChannel = (SocketChannel) key.channel();

        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();) {
            int size=0;
            for (; (size = socketChannel.read(byteBuffer)) > 0;){
                byteBuffer.flip();
                baos.write(byteBuffer.array(), 0, size);
                byteBuffer.clear();
            }
            final byte[] content = baos.toByteArray();

            System.out.println("receive data: "+new String(content));

            ByteBuffer writeBuf = ByteBuffer.allocate(content.length);
            writeBuf.put(content);
            writeBuf.flip();// 是将指针重新指到0位置
            socketChannel.write(writeBuf);

            if (size == -1){
                socketChannel.close();
            } else {
                // 读完之后，还要关心读，才能连续不断地读
                key.interestOps(key.interestOps()|SelectionKey.OP_READ);
                // 通知大管家执行
                key.selector().wakeup();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
